package com.leyantech.ecp.agenx.connector;

import com.leyantech.chaos.kafka.KafkaProducerStringClient;
import com.leyantech.ecp.agenx.common.ClusterVerticle;
import com.leyantech.ecp.agenx.common.config.Constants;
import com.leyantech.ecp.agenx.common.di.JWTAuthProvider;
import com.leyantech.ecp.agenx.common.misc.HttpLocalRateLimiter;
import com.leyantech.ecp.agenx.common.misc.TracingHelper;
import com.leyantech.ecp.agenx.connector.di.ConnectorModule;
import com.leyantech.ecp.agenx.connector.handler.WebSocketConnectHandler;

import com.google.common.collect.Lists;
import com.google.inject.Inject;
import com.google.inject.Module;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.prometheus.client.vertx.MetricsHandler;
import io.vertx.core.http.HttpServer;
import io.vertx.ext.auth.User;
import io.vertx.ext.healthchecks.HealthCheckHandler;
import io.vertx.ext.web.Router;
import io.vertx.ext.web.handler.LoggerHandler;
import org.apache.commons.lang3.StringUtils;

import java.util.List;

/**
 * @author eahau, {@literal <eahau@foxmail.com>}
 * @date 2022-10-25.
 */
public class ConnectorVerticle extends ClusterVerticle {

  @Inject
  private WebSocketConnectHandler webSocketConnectHandler;

  @Inject
  private KafkaProducerStringClient kafkaClient;

  @Inject
  private JWTAuthProvider jwtAuthProvider;

  @Override
  protected boolean isLiteMember() {
    return false;
  }

  @Override
  protected List<Module> dependentModules() {
    return Lists.newArrayList(new ConnectorModule());
  }

  @Override
  public void start() {

    final Router router = Router.router(vertx);

    router.route()
        .handler(TracingHelper.tracingHandler(LoggerHandler.create()))
        .failureHandler(TracingHelper.tracingHandler(context -> {
          if (context.failed()) {
            final Throwable failure = context.failure();
            final int code = context.statusCode();
            log.error("handle failed, code {}", code, failure);
            context.response()
                .setStatusCode(code)
                .end(failure.getMessage());
            return;
          }
          context.end();
        }));

    router.route("/").handler(context -> context.end("hello world."));

    configHealthCheck(healthChecks -> {
          router.route("/health*")
              .blockingHandler(HealthCheckHandler.createWithHealthChecks(healthChecks));
        }
    );

    router.get("/metrics").blockingHandler(new MetricsHandler());

    router.get("/chat")
        .handler(
            new HttpLocalRateLimiter(context -> context.request().remoteAddress().hostAddress()))
        // 单台实例，每秒新建连接数限流
        .handler(new HttpLocalRateLimiter(context -> "websocket"))
        .handler(context -> jwtAuthProvider.getJwtAuthHandler().handle(context))
        .handler(TracingHelper.tracingHandler(context -> {
          final User user = context.user();
          final String storeId, assistantId;
          if (user == null || StringUtils.isAnyBlank(
              storeId = user.get(Constants.STORE_ID_FIELD),
              assistantId = user.get(Constants.ASSISTANT_ID_FIELD))) {
            context.fail(HttpResponseStatus.UNAUTHORIZED.code());
            return;
          }

          log.info("{} {} connecting...", storeId, assistantId);
          context.next();
        }))
        .handler(TracingHelper.tracingHandler(webSocketConnectHandler));

    final HttpServer httpServer = vertx.createHttpServer();
    httpServer.requestHandler(router)
        .listen(8080);

    // hook
    addCloseHook(httpServer::close);
    addRunnableCloseHook(kafkaClient::destroy);
  }

  public static void main(String[] args) {
    new ConnectorVerticle().startVerticle();
  }

}
